package com.aiwiown.snackmq.console.service;

import com.aiwiown.snackmq.api.Producer;
import com.aiwiown.snackmq.api.config.ProducerConfig;
import com.aiwiown.snackmq.api.exception.SnackMQClientException;
import com.aiwiown.snackmq.api.impl.DefaultProducer;
import com.aiwiown.snackmq.common.auth.AuthenticatedUser;
import com.aiwiown.snackmq.common.auth.Permission;
import com.aiwiown.snackmq.common.auth.User;
import com.aiwiown.snackmq.common.dto.TopicInfo;
import com.aiwiown.snackmq.common.dto.console.BrokerInfoDTO;
import com.aiwiown.snackmq.common.dto.console.ClusterOverviewDTO;
import com.aiwiown.snackmq.common.dto.console.ConsumerGroupDTO;
import com.aiwiown.snackmq.common.dto.console.ExchangeInfoDTO;
import com.aiwiown.snackmq.common.dto.console.RoleInfoDTO;
import com.aiwiown.snackmq.common.message.Message;
import com.aiwiown.snackmq.common.message.MessageStatus;
import com.aiwiown.snackmq.common.message.MessageType;
import com.aiwiown.snackmq.common.routing.ExchangeType;
import com.aiwiown.snackmq.common.serialization.JsonSerializer;
import com.aiwiown.snackmq.common.serialization.SerializerFactory;
import com.aiwiown.snackmq.console.config.SnackMQAdminConfig;
import com.google.common.base.Strings;
import com.google.common.reflect.TypeToken;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.servlet.http.HttpSession;
import java.lang.reflect.Type;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
@Service
public class SnackMQAdminService {

    private final JsonSerializer serializer = SerializerFactory.getJsonSerializer();

    @Autowired
    private SnackMQAdminConfig config;
    @Autowired
    private HttpSession httpSession;

    private Producer adminProducer;

    // --- 【新增】用于自动重连的组件 ---
    private ScheduledExecutorService healthCheckScheduler;
    private final AtomicBoolean isReconnecting = new AtomicBoolean(false);


    @PostConstruct
    public void init() {
        // 【改造】将初始化逻辑与定时健康检查启动相结合
        log.info("Starting SnackMQAdminService...");
        initializeProducer();
        startHealthCheckScheduler();
    }

    @PreDestroy
    public void destroy() {
        // 【改造】确保关闭生产者和定时器
        log.info("Stopping SnackMQAdminService...");
        if (healthCheckScheduler != null && !healthCheckScheduler.isShutdown()) {
            healthCheckScheduler.shutdownNow();
            log.info("Admin client health check scheduler stopped.");
        }
        shutdownProducer();
    }

    /**
     * 【新增】初始化生产者实例。
     */
    private void initializeProducer() {
        log.info("Initializing SnackMQ Admin Producer to connect to {}", config.getBrokerAddress());
        ProducerConfig producerConfig = ProducerConfig.builder()
                .username(config.getUsername())
                .password(config.getPassword())
                .build();
        this.adminProducer = new DefaultProducer(config.getBrokerAddress(), producerConfig);
        try {
            this.adminProducer.start();
            log.info("SnackMQ admin producer started and connected successfully.");
        } catch (SnackMQClientException e) {
            // 在启动时，连接失败是正常的，健康检查器会处理后续的重连。
            log.error("Failed to connect SnackMQ admin producer on initial startup. Will retry via health checks.", e);
        }
    }

    /**
     * 【新增】安全地关闭当前的生产者实例。
     */
    private void shutdownProducer() {
        if (this.adminProducer != null) {
            try {
                this.adminProducer.close();
                log.info("SnackMQ admin producer instance shut down.");
            } catch (Exception e) {
                log.error("Error shutting down SnackMQ admin producer.", e);
            }
        }
    }

    /**
     * 【新增】启动一个定时任务，定期检查与 Broker 的连接健康状况。
     */
    private void startHealthCheckScheduler() {
        healthCheckScheduler = Executors.newSingleThreadScheduledExecutor(r -> new Thread(r, "snackmq-admin-health-checker"));
        // 【修复】将首次健康检查的延迟从30秒缩短到5秒，以更快地从启动时的竞态条件中恢复。
        // 后续检查仍保持30秒的间隔，以避免不必要的系统负载。
        healthCheckScheduler.scheduleAtFixedRate(this::healthCheck, 5, 30, TimeUnit.SECONDS);
        log.info("Started admin client health check scheduler.");
    }

    /**
     * 【新增】健康检查的核心逻辑。
     */
    private void healthCheck() {
        log.debug("Performing admin client health check...");
        if (adminProducer == null || !adminProducer.isActive()) {
            log.warn("Health check determined the admin client is disconnected. Triggering reconnect.");
            reconnect();
        }
    }

    /**
     * 【新增】执行重连的核心方法，此方法是线程安全的。
     */
    private void reconnect() {
        if (isReconnecting.compareAndSet(false, true)) {
            log.info("Reconnection process started...");
            try {
                shutdownProducer();
                initializeProducer();
                log.info("Reconnection process completed.");
            } finally {
                isReconnecting.set(false);
            }
        } else {
            log.info("Reconnection is already in progress.");
        }
    }

    /**
     * 登录并获取认证令牌。
     *
     * @param username 用户名
     * @param password 密码
     * @return 一个包含 JWT 的 AuthenticatedUser 对象
     * @throws SnackMQClientException 如果认证失败
     */
    public AuthenticatedUser login(String username, String password) throws SnackMQClientException {
        if (adminProducer == null) {
            throw new IllegalStateException("Admin producer is not initialized or connected. Please wait for reconnection.");
        }
        // 假设您的 Producer 接口和实现中已有 authenticate 方法
        return adminProducer.authenticate(username, password);
    }

    /**
     * 获取集群概览信息。
     *
     * @return 集群概览 DTO
     * @throws SnackMQClientException 如果请求失败
     */
    public ClusterOverviewDTO getClusterOverview() throws SnackMQClientException {
        Message request = Message.builder()
                .type(MessageType.GET_CLUSTER_OVERVIEW_REQUEST)
                .build();

        Message response = sendAdminRequest(request);
        return serializer.deserialize(response.getBody(), ClusterOverviewDTO.class);
    }

    /**
     * 【新增】获取所有 Topic 的信息。
     */
    public List<TopicInfo> listTopics() throws SnackMQClientException {
        Message request = Message.builder()
                .type(MessageType.LIST_TOPICS_REQUEST)
                .build();

        Message response = sendAdminRequest(request);
        Type listType = new TypeToken<ArrayList<TopicInfo>>() {
        }.getType();
        return serializer.deserialize(response.getBody(), listType);
    }

    /**
     * 【新增】获取所有消费者组的信息。
     */
    public List<ConsumerGroupDTO> listConsumerGroups() throws SnackMQClientException {
        Message request = Message.builder()
                .type(MessageType.LIST_CONSUMER_GROUPS_REQUEST)
                .build();

        Message response = sendAdminRequest(request);
        Type listType = new TypeToken<ArrayList<ConsumerGroupDTO>>() {
        }.getType();
        return serializer.deserialize(response.getBody(), listType);
    }

    /**
     * 【新增】获取单个 Topic 的详细信息。
     *
     * @param topicName 主题名称
     * @return Topic 详情
     * @throws SnackMQClientException 如果获取失败
     */
    public TopicInfo getTopicDetails(String topicName) throws SnackMQClientException {
        if (adminProducer == null) {
            throw new SnackMQClientException("Admin producer is not connected.");
        }
        // 直接调用 client 模块提供的能力
        return adminProducer.getTopicMetadata(topicName);
    }

    /**
     * 【新增】创建一个新的 Topic。
     *
     * @param topicName         主题名称
     * @param partitions        分区数
     * @param replicationFactor 副本因子
     * @throws SnackMQClientException 如果创建失败
     */
    public void createTopic(String topicName, int partitions, int replicationFactor) throws SnackMQClientException {
        if (adminProducer == null) {
            throw new SnackMQClientException("Admin producer is not connected.");
        }
        // 直接调用 client 模块提供的能力
        adminProducer.createTopic(topicName, partitions, replicationFactor);
    }


    /**
     * 【新增】删除一个 Topic。
     *
     * @param topicName 待删除的 Topic 名称
     */
    public void deleteTopic(String topicName) throws SnackMQClientException {
        if (adminProducer == null) {
            throw new SnackMQClientException("Admin producer is not connected.");
        }
        // 直接调用 client 模块提供的能力
        adminProducer.deleteTopic(topicName);
    }

    /**
     * 【新增】声明一个交换机。
     *
     * @param exchangeName 交换机名称
     * @param type         交换机类型
     * @throws SnackMQClientException 如果声明失败
     */
    public void declareExchange(String exchangeName, ExchangeType type) throws SnackMQClientException {
        if (adminProducer == null) {
            throw new SnackMQClientException("Admin producer is not connected.");
        }
        adminProducer.declareExchange(exchangeName, type);
    }


    /**
     * 【新增】删除一个交换机。
     *
     * @param exchangeName 待删除的交换机名称
     * @throws SnackMQClientException 如果删除失败
     */
    public void deleteExchange(String exchangeName) throws SnackMQClientException {
        if (adminProducer == null) {
            throw new SnackMQClientException("Admin producer is not connected.");
        }
        adminProducer.deleteExchange(exchangeName);
    }

    /**
     * 【新增】将一个队列（主题）绑定到交换机。
     *
     * @param queueName    队列/主题名称
     * @param exchangeName 交换机名称
     * @param routingKey   路由键
     * @throws SnackMQClientException 如果绑定失败
     */
    public void bindQueue(String queueName, String exchangeName, String routingKey) throws SnackMQClientException {
        if (adminProducer == null) {
            throw new SnackMQClientException("Admin producer is not connected.");
        }
        adminProducer.bindQueue(queueName, exchangeName, routingKey);
    }

    /**
     * 【新增】获取所有用户列表。
     */
    public List<User> listUsers() throws SnackMQClientException {
        Message request = Message.builder().type(MessageType.LIST_USERS_REQUEST).build();
        Message response = sendAdminRequest(request);
        Type listType = new TypeToken<ArrayList<User>>() {
        }.getType();
        return serializer.deserialize(response.getBody(), listType);
    }

    /**
     * 【新增】获取所有角色及其权限。
     */
    public List<RoleInfoDTO> listRoles() throws SnackMQClientException {
        Message request = Message.builder().type(MessageType.LIST_ROLES_REQUEST).build();
        Message response = sendAdminRequest(request);
        Type listType = new TypeToken<ArrayList<RoleInfoDTO>>() {
        }.getType();
        return serializer.deserialize(response.getBody(), listType);
    }

    /**
     * 【新增】调用客户端 API 创建一个新用户。
     */
    public void createUser(String username, String password, List<String> roles) throws SnackMQClientException {
        if (adminProducer == null) {
            throw new SnackMQClientException("Admin producer is not connected.");
        }
        adminProducer.createUser(username, password, roles);
    }

    /**
     * 【新增】调用客户端 API 设置角色权限。
     */
    public void setRolePermissions(String roleName, List<Permission> permissions) throws SnackMQClientException {
        if (adminProducer == null) {
            throw new SnackMQClientException("Admin producer is not connected.");
        }
        adminProducer.setRolePermissions(roleName, permissions);
    }

    /**
     * 【新增】调用客户端 API 删除一个用户。
     */
    public void deleteUser(String username) throws SnackMQClientException {
        if (adminProducer == null) {
            throw new SnackMQClientException("Admin producer is not connected.");
        }
        adminProducer.deleteUser(username);
    }

    /**
     * 【新增】调用客户端 API 删除一个角色。
     */
    public void deleteRole(String roleName) throws SnackMQClientException {
        if (adminProducer == null) {
            throw new SnackMQClientException("Admin producer is not connected.");
        }
        adminProducer.deleteRole(roleName);
    }

    /**
     * 【新增】发送一条测试消息。
     *
     * @param destination 目的地（Topic 或 Exchange 名称）
     * @param isExchange  目的地是否为 Exchange
     * @param routingKey  路由键（用于 Exchange）
     * @param messageKey  消息的 Key
     * @param messageBody 消息体
     * @throws SnackMQClientException 如果发送失败
     */
    public void sendTestMessage(String destination, boolean isExchange, String routingKey, String messageKey, String messageBody) throws SnackMQClientException {
        if (adminProducer == null) {
            throw new SnackMQClientException("Admin producer is not connected.");
        }

        Message.MessageBuilder messageBuilder = Message.builder()
                .key(messageKey)
                .body(messageBody.getBytes(StandardCharsets.UTF_8));

        if (isExchange) {
            // 当目标是 Exchange 时，将 Exchange 名称放入消息属性中，
            // Broker 的消息处理器将据此进行路由。
            messageBuilder.property("exchangeName", destination);
            // 对于 Topic 和 Direct 类型的 Exchange，路由键至关重要。
            messageBuilder.topic(routingKey);
        } else {
            // 当目标是 Topic 时，直接设置消息的主题。
            messageBuilder.topic(destination);
        }

        // 使用 admin producer 发送构建好的消息。
        adminProducer.send(messageBuilder.build());
    }

    /**
     * 【新增】按位点查询单条消息。
     *
     * @param topic     主题
     * @param partition 分区
     * @param offset    位点
     * @return 查询到的消息，如果不存在则返回 null
     * @throws SnackMQClientException 如果查询失败
     */
    public Message peekMessage(String topic, int partition, long offset) throws SnackMQClientException {
        if (adminProducer == null) {
            throw new SnackMQClientException("Admin producer is not connected.");
        }

        Message request = Message.builder()
                .type(MessageType.PULL_REQUEST)
                .topic(topic)
                .property("partition", String.valueOf(partition))
                .property("offset", String.valueOf(offset))
                .property("maxMessages", "1")
                // 为 peek 操作指定一个临时的、专用的消费者组名
                .property("consumerGroup", "snackmq-console-peek-tool")
                .build();

        // --- 【核心修复】: 不再使用通用的 sendAdminRequest 方法，而是自定义请求处理逻辑 ---
        // 这是因为 "peek" 操作的 "NOT_FOUND" 或 "EMPTY" 状态是预期的正常结果，而不应被视为错误。
        // 通用的 sendAdminRequest 方法会把所有非 SUCCESS 状态都当作异常抛出。
        try {
            Message authenticatedRequest = buildAuthenticatedRequest(request);
            // 【修复】调用 DefaultProducer 中新的、公开的、能处理数据平面重定向的方法
            // 旧的 sendRequestWithRedirectHandling 方法已被移除
            CompletableFuture<Message> responseFuture = ((DefaultProducer) adminProducer).sendRequestToPartitionLeader(authenticatedRequest, topic, partition);
            Message response = responseFuture.get(10, TimeUnit.SECONDS);

            // 检查响应状态
            if (response.getStatus() == MessageStatus.SUCCESS) {
                // 成功找到消息
                return (response.getMessages() != null && !response.getMessages().isEmpty())
                        ? response.getMessages().get(0) : null;
            } else if (response.getStatus() == MessageStatus.NOT_FOUND || response.getStatus() == MessageStatus.EMPTY) {
                // 未找到消息，这是正常情况，返回 null
                return null;
            } else {
                // 其他失败状态，则抛出异常
                String errorMessage = response.getErrorMessage() != null ? response.getErrorMessage() : "Unknown error from broker.";
                throw new SnackMQClientException("Peek message command failed: " + errorMessage);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new SnackMQClientException("Peek message command was interrupted.", e);
        } catch (ExecutionException e) {
            throw new SnackMQClientException("Peek message command execution failed.", e.getCause());
        } catch (TimeoutException e) {
            throw new SnackMQClientException("Peek message command timed out.", e);
        }
    }

    /**
     * 【新增】列出所有 Exchange。
     *
     * @return 包含所有 Exchange 的列表
     * @throws SnackMQClientException 如果列出 Exchange 失败
     */
    public List<ExchangeInfoDTO> listExchanges() throws SnackMQClientException {
        Message request = Message.builder()
                .type(MessageType.LIST_EXCHANGES_REQUEST)
                .build();
        Message response = sendAdminRequest(request);
        Type listType = new TypeToken<ArrayList<ExchangeInfoDTO>>() {
        }.getType();
        return serializer.deserialize(response.getBody(), listType);
    }

    /**
     * 【新增】诊断分区状态，分析段文件和索引信息。
     *
     * @param topic     主题名称
     * @param partition 分区ID
     * @return 诊断结果
     * @throws SnackMQClientException 如果诊断失败
     */
    public Map<String, Object> diagnosePartition(String topic, int partition) throws SnackMQClientException {
        if (adminProducer == null) {
            throw new SnackMQClientException("Admin producer is not connected.");
        }

        Message request = Message.builder()
                .type(MessageType.DIAGNOSE_PARTITION_REQUEST)
                .topic(topic)
                .property("partition", String.valueOf(partition))
                .build();

        try {
            Message authenticatedRequest = buildAuthenticatedRequest(request);
            Message response = ((DefaultProducer) adminProducer).executeWithRedirectRetry(authenticatedRequest, "Diagnose partition");
            
            Type mapType = new TypeToken<Map<String, Object>>() {}.getType();
            return serializer.deserialize(response.getBody(), mapType);
        } catch (Exception e) {
            if (e instanceof SnackMQClientException) {
                throw e;
            }
            throw new SnackMQClientException("Diagnose partition command execution failed.", e);
        }
    }

    /**
     * 【新增】修复分区损坏的段。
     *
     * @param topic     主题名称
     * @param partition 分区ID
     * @throws SnackMQClientException 如果修复失败
     */
    public void repairPartition(String topic, int partition) throws SnackMQClientException {
        if (adminProducer == null) {
            throw new SnackMQClientException("Admin producer is not connected.");
        }

        Message request = Message.builder()
                .type(MessageType.REPAIR_PARTITION_REQUEST)
                .topic(topic)
                .property("partition", String.valueOf(partition))
                .build();

        try {
            Message authenticatedRequest = buildAuthenticatedRequest(request);
            ((DefaultProducer) adminProducer).executeWithRedirectRetry(authenticatedRequest, "Repair partition");
        } catch (Exception e) {
            if (e instanceof SnackMQClientException) {
                throw e;
            }
            throw new SnackMQClientException("Repair partition command execution failed.", e);
        }
    }

    /**
     * 【新增】手动修复分区段结构。
     *
     * @param topic     主题名称
     * @param partition 分区ID
     * @throws SnackMQClientException 如果修复失败
     */
    public void manualRepairPartition(String topic, int partition) throws SnackMQClientException {
        if (adminProducer == null) {
            throw new SnackMQClientException("Admin producer is not connected.");
        }

        Message request = Message.builder()
                .type(MessageType.MANUAL_REPAIR_PARTITION_REQUEST)
                .topic(topic)
                .property("partition", String.valueOf(partition))
                .build();

        try {
            Message authenticatedRequest = buildAuthenticatedRequest(request);
            ((DefaultProducer) adminProducer).executeWithRedirectRetry(authenticatedRequest, "Manual repair partition");
        } catch (Exception e) {
            if (e instanceof SnackMQClientException) {
                throw e;
            }
            throw new SnackMQClientException("Manual repair partition command execution failed.", e);
        }
    }

    /**
     * 【新增】解绑队列。
     *
     * @param queueName    队列名称
     * @param exchangeName 交换机名称
     * @param routingKey   路由键
     * @throws SnackMQClientException 如果解绑队列失败
     */
    public void unbindQueue(String queueName, String exchangeName, String routingKey) throws SnackMQClientException {
        if (adminProducer == null) {
            throw new SnackMQClientException("Admin producer is not connected.");
        }
        // 调用在 Producer 接口中新定义的方法
        adminProducer.unbindQueue(queueName, exchangeName, routingKey);
    }

    /**
     * 发送管理请求并同步等待响应的私有辅助方法。
     *
     * @param request 要发送的管理请求消息
     * @return Broker 的响应消息
     * @throws SnackMQClientException 如果请求失败、超时或 Broker 返回错误
     */
    private Message sendAdminRequest(Message request) throws SnackMQClientException {
        if (adminProducer == null) {
            throw new SnackMQClientException("Admin producer is not connected. Please try again shortly.");
        }
        try {
            // 【核心修复】在发送请求前，总是尝试附加当前用户的认证令牌。
            // 如果用户未登录（session中没有token），则会依赖 adminProducer 内部的令牌。
            Message authenticatedRequest = buildAuthenticatedRequest(request);

            // 【修复】使用 executeWithRedirectRetry 来处理重定向，而不是直接使用 sendRequestWithRedirectHandling
            return ((DefaultProducer) adminProducer).executeWithRedirectRetry(authenticatedRequest, "Admin command");

        } catch (Exception e) {
            // 如果 executeWithRedirectRetry 失败，包装异常
            if (e instanceof SnackMQClientException) {
                throw e;
            }
            throw new SnackMQClientException("Admin command execution failed.", e);
        }
    }

    private Message buildAuthenticatedRequest(Message request) {
        if (httpSession == null) {
            return request;
        }
        String token = (String) httpSession.getAttribute("user_token");
        if (!Strings.isNullOrEmpty(token)) {
            // 使用 toBuilder() 模式，在不改变原始请求对象的情况下创建一个新实例
            return request.toBuilder()
                    .property("authToken", token)
                    .build();
        }
        // 如果没有 token，返回原始请求
        return request;
    }

    /**
     * 获取集群中所有 Broker 节点的信息。
     *
     * @return Broker 信息列表
     * @throws SnackMQClientException 如果请求失败
     */
    public List<BrokerInfoDTO> listBrokers() throws SnackMQClientException {
        Message request = Message.builder()
                .type(MessageType.LIST_BROKERS_REQUEST)
                .build();

        Message response = sendAdminRequest(request);
        // 使用 TypeToken 来处理泛型列表的反序列化
        Type listType = new TypeToken<ArrayList<BrokerInfoDTO>>() {
        }.getType();
        return serializer.deserialize(response.getBody(), listType);
    }

}